package org.groupg.project.dagdemo;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

public class DisruptorExample {
    public static void main(String[] args) throws Exception {
        // Executor that will run the event handlers
        ExecutorService executor = Executors.newCachedThreadPool();

        ThreadFactory customThreadFactory = new ThreadFactory() {
            private final AtomicInteger threadNumber = new AtomicInteger(1);

            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("CustomThread-" + threadNumber.getAndIncrement());
                return thread;
            }
        };
        // The factory for the stage 1 event
        Stage1EventFactory stage1EventFactory = new Stage1EventFactory();

        // Specify the size of the ring buffer, must be power of 2
        int bufferSize = 1024;

        // Construct the disruptor for stage 1
        Disruptor<Stage1Event> stage1Disruptor = new Disruptor<>(
                stage1EventFactory,
                bufferSize,
                customThreadFactory,
                ProducerType.SINGLE,
                new YieldingWaitStrategy());

        // Connect the handler for stage 1
        stage1Disruptor.handleEventsWith(new Stage1EventHandler(null)); // Placeholder, will be set later

        // Create a ring buffer for stage 1
        RingBuffer<Stage1Event> stage1RingBuffer = stage1Disruptor.start();

        // The factory for the stage 2 event
        Stage2EventFactory stage2EventFactory = new Stage2EventFactory();

        // Construct the disruptor for stage 2
        Disruptor<Stage2Event> stage2Disruptor = new Disruptor<>(
                stage2EventFactory,
                bufferSize,
                customThreadFactory,
                ProducerType.SINGLE,
                new YieldingWaitStrategy());

        // Connect the handler for stage 2
        stage2Disruptor.handleEventsWith(new Stage2EventHandler());

        // Create a ring buffer for stage 2
        RingBuffer<Stage2Event> stage2RingBuffer = stage2Disruptor.start();

        // Set the stage2RingBuffer in the stage1 event handler
        stage1Disruptor.handleEventsWith(new Stage1EventHandler(stage2RingBuffer));

        // Example: Publish an event to stage 1
        long sequence = stage1RingBuffer.next();
        try {
            Stage1Event event = stage1RingBuffer.get(sequence);
            event.setValue(42);
        } finally {
            stage1RingBuffer.publish(sequence);
        }

        // Shutdown the Disruptor and executor
        stage1Disruptor.shutdown();
        stage2Disruptor.shutdown();
        executor.shutdown();
    }
}
